package com.sugon.ohdfs.integration.flink.service;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;

public class TestItemSource implements SourceFunction<TestItem> {

    String format = "yyyy-MM-dd HH:mm:ss";
    boolean shutdown = false;

    @Override
    public void run(SourceContext<TestItem> ctx) throws Exception {
        while (!shutdown){
            LocalDateTime timestamp = LocalDateTime.now();
            ctx.collectWithTimestamp(TestItem.builder()
                            .name("test")
                            .value(DateTimeFormatter.ofPattern(format).format(timestamp))
                            .produceTimestamp(timestamp)
                            .build(),
                    timestamp.toEpochSecond(ZoneOffset.systemDefault().getRules().getOffset(timestamp)));
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        shutdown = true;
    }
}
